package cn.itcast.streaming.window.function;

import cn.itcast.streaming.entity.ItcastDataObj;
import com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;

/**
 * Author itcast
 * Date 2021/9/24 11:32
 * 实现15分钟生成一个行程数据，分析行程采样分析统计
 * 开发步骤：
 * 1.重写 apply 方法
 * 1.1 将迭代器转换成集合列表
 * 1.2 对集合列表的数据进行排序操作
 * 1.3 首先要获取排序后的第一条数据从而得到周期（5m等）的开始时间
 * 1.4 采样数据的soc(剩余电量百分比)、mileage（累计里程）、speed（车速）、gps（经度+维度）、terminalTime（终端时间）字段属性需要拼接到一个字符串返回
 * soc(剩余电量百分比)
 * mileage（累计里程）
 * speed（车速）
 * gps：地理位置
 * terminalTime：终端时间
 * 1.5 获取排序后的最后一条数据作为当前窗口的最后一条数据
 * 1.6 获取窗口的第一条数据的终端时间作为开始时间戳
 * 1.7 获取窗口的最后一条数据的终端时间作为结束时间戳
 * 1.8 遍历窗口内的每条数据，计算5s,10s,15s,20s 采样周期内的数据
 * 1.9 创建字符串数组类型用于存储采集到的车辆唯一编码，终端时间戳，剩余电量，总里程数，车速，地理gps，终端时间
 * 1.10 返回数据
 */
public class DriveSampleWindowFunction extends RichWindowFunction<ItcastDataObj, String[], String, TimeWindow> {

    //1.重写 apply 方法
    @Override
    public void apply(String key, TimeWindow window, Iterable<ItcastDataObj> input, Collector<String[]> out) throws Exception {
        //1.1 将迭代器转换成集合列表,数据有延迟
        ArrayList<ItcastDataObj> itcastDataObjs = Lists.newArrayList(input);
        //1.2 对集合列表的数据进行排序操作,升序排序
        Collections.sort(
                itcastDataObjs,
                new Comparator<ItcastDataObj>() {
                    @Override
                    public int compare(ItcastDataObj o1, ItcastDataObj o2) {
                        if (o1.getTerminalTimeStamp() > o2.getTerminalTimeStamp()) {
                            return 1;
                        } else if (o1.getTerminalTimeStamp() < o2.getTerminalTimeStamp()) {
                            return -1;
                        } else {
                            return 0;
                        }
                    }
                });
        //1.3 首先要获取排序后的第一条数据从而得到周期（5s等）的开始时间
        ItcastDataObj firstItcastData = itcastDataObjs.get(0);
        //1.4 采样数据的soc(剩余电量百分比)、mileage（累计里程）、speed（车速）、gps（经度+维度）、terminalTime（终端时间）字段属性需要拼接到一个字符串返回
        //soc(剩余电量百分比)
        StringBuffer socBuffer = new StringBuffer(firstItcastData.getSoc());
        //mileage（累计里程）
        StringBuffer mileageBuffer = new StringBuffer(firstItcastData.getTotalOdometer()+"");
        //speed（车速）
        StringBuffer speedBuffer = new StringBuffer(firstItcastData.getSpeed()+"");
        //gps：地理位置
        StringBuffer gpsBuffer = new StringBuffer(firstItcastData.getLng()+"|"+firstItcastData.getLat());
        //terminalTime：终端时间
        StringBuffer terminalTimeBuffer = new StringBuffer(firstItcastData.getTerminalTime());
        //1.5 获取排序后的最后一条数据作为当前窗口的最后一条数据
        ItcastDataObj lastItcastDataObj = itcastDataObjs.get(itcastDataObjs.size() - 1);
        //1.6 获取窗口的第一条数据的终端时间作为开始时间戳
        Long startTimestamp = firstItcastData.getTerminalTimeStamp();
        //1.7 获取窗口的最后一条数据的终端时间作为结束时间戳
        Long lastTimestamp = lastItcastDataObj.getTerminalTimeStamp();
        //1.8 遍历窗口内的每条数据，计算5s,10s,15s,20s 采样周期内的数据
        for (ItcastDataObj itcastDataObj : itcastDataObjs) {
            Long currentTimestamp = itcastDataObj.getTerminalTimeStamp();
            if (currentTimestamp - startTimestamp >= 5 * 1000 || currentTimestamp == lastTimestamp) {
                socBuffer.append("," + itcastDataObj.getSoc());
                mileageBuffer.append("," + itcastDataObj.getTotalOdometer());
                speedBuffer.append("," + itcastDataObj.getSpeed());
                gpsBuffer.append("," + itcastDataObj.getLng() + "|" + itcastDataObj.getLat());
                terminalTimeBuffer.append("," + itcastDataObj.getTerminalTime());
                startTimestamp = currentTimestamp;
            }
        }
        //1.9 创建字符串数组类型用于存储采集到的
        // 车辆唯一编码，终端时间戳，剩余电量，总里程数，车速，地理gps，终端时间
        String[] sampleArry = new String[7];
        sampleArry[0] = firstItcastData.getVin();
        sampleArry[1] = firstItcastData.getTerminalTimeStamp()+"";
        sampleArry[2] = socBuffer.toString();
        sampleArry[3] = mileageBuffer.toString();
        sampleArry[4] = speedBuffer.toString();
        sampleArry[5] = gpsBuffer.toString();
        sampleArry[6] = terminalTimeBuffer.toString();

        //1.10 返回数据
        out.collect(sampleArry);
    }
}
